package com.sgy.rocketMQ.producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sgy.netty.handler.UserInfoManager;
import com.sgy.pojo.UserInfo;
import com.sgy.rocketMQ.OfflineInfoTransmit;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Component
public class OfflineInfoHelper implements OfflineInfoTransmit {

    private DefaultMQProducer producer;
    @Value("${rocket.producer.namesrvAddr}")
    private String namesrvAddr;
    private ConcurrentMap<String, UserInfo> userInfos = UserInfoManager.userChannelMap;
    private ConcurrentMap<Channel,DefaultMQPushConsumer> consumers = new ConcurrentHashMap<>();
    private Map map = new HashMap();

    public OfflineInfoHelper() {
        producer = new DefaultMQProducer("send");
        producer.setNamesrvAddr("10.10.4.64:9876");
        producer.setVipChannelEnabled(false);
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    //顺序消息，将消息存储到消息队列，生产者
    @Override
    public void pushP2P(Integer uid, JSONObject data) {
        try {
            Message msg = new Message("p2p_"+uid,data.toString().getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                //selector队列选择器，发送时会回调
                // mqs是队列集合，也就是topic所对应的所有队列
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Integer id = (Integer) arg;
                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            },uid);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //注册用户推送消息的通道
    @Override
    public void registerPull(Channel channel,Integer uid) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_"+uid);
        try {
            consumer.subscribe("p2p_"+uid,"*");
            consumer.setNamesrvAddr("10.10.4.64:9876");
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //消费线程池最大数量
            consumer.setConsumeThreadMax(1);
            //消费线程池最小数量
            consumer.setConsumeThreadMin(1);
            //设置一个Listener，主要进行消息的逻辑处理
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
//                        map.put("username",username);
//                        map.put("mess",msg.getBody());
                        System.out.println(Thread.currentThread().getName() + "consumer:" + new String(msg.getBody()));
                        channel.writeAndFlush(new TextWebSocketFrame(new String(msg.getBody())));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            consumers.put(channel,consumer);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    //注销消息队列通道
    @Override
    public void unregisterPull(Channel channel) {
        DefaultMQPushConsumer consumer = consumers.get(channel);
        consumer.shutdown();
        consumers.remove(channel);
    }

}
